package com.basic.demo;

import com.basic.pojo.UserInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @author xl-9527
 * @since 2024/11/29
 **/
public class SimpleDemo {

    public void executeSocket() throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        executeSocket(executionEnvironment);

        executionEnvironment.execute();
    }

    public void executeKafka() throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        executeKafka(executionEnvironment);

        executionEnvironment.execute();
    }

    private void executeKafka(final StreamExecutionEnvironment executionEnvironment) {
        executionEnvironment.enableCheckpointing(2000, CheckpointingMode.AT_LEAST_ONCE);
        KafkaSource<UserInfo> kafkaSource = KafkaSource.<UserInfo>builder().setBootstrapServers("192.168.4.81:9092")
                .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(UserInfo.class))
                .setTopics("flink-kafka")
                .build();
        executionEnvironment.setParallelism(Runtime.getRuntime().availableProcessors())
                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source")
                .returns(Types.POJO(UserInfo.class)).map(UserInfo::getAge).returns(Types.INT).print();
    }

    /**
     * socket 的示例计算每个字符的数量
     */
    private static void executeSocket(final StreamExecutionEnvironment executionEnvironment) {
        executionEnvironment.setParallelism(Runtime.getRuntime().availableProcessors())
                .socketTextStream("192.168.4.11", 8899).flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (socketVal, out) -> {
                    if (socketVal != null && !socketVal.isEmpty()) {
                        for (final char c : socketVal.toCharArray()) {
                            out.collect(Tuple2.of(String.valueOf(c), 1));
                        }
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(v -> v.f0).sum(1)
                .print();
    }
}
